package com.ada.spark.accumulator

import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 累加器用来对信息进行聚合，通常在向 Spark传递函数时，
  * 比如使用 map() 函数或者用 filter() 传条件时，可以使用驱动器程序中定义的变量，
  * 但是集群中运行的每个任务都会得到这些变量的一份新的副本，更新这些副本的值也不会影响驱动器中的对应变量。
  * 如果我们想实现所有分片处理时更新共享变量的功能，那么累加器可以实现我们想要的效果。
  */
object AccumulatorTest {
    def main(args: Array[String]): Unit = {
        //获取Spark配置信息并创建与spark的连接
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("AccumulatorTest")
        val sc = new SparkContext(sparkConf)

        val notice = sc.textFile("in")

        //创建累加器
        val blanklines = new LongAccumulator();
        //注册累加器
        sc.register(blanklines)

        val tmp = notice.flatMap(line => {
            if (line == "") {
                blanklines.add(1)
            }
            line.split(" ")
        })

        println(tmp.count()) //17

        println(blanklines.value) //5

    }
}
